package org.apache.rocketmq.remoting.netty;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.concurrent.Future;
import io.opentelemetry.api.common.AttributesBuilder;
import org.apache.rocketmq.common.*;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;

import javax.annotation.Nullable;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2024/12/25
 * @方法描述：Netty服务器的抽象父类，在第一版本代码中，这个类的内容非常简单，在第二版本代码中，这个类就会被填充非常多的内容，实际上netty服务端处理接收到的请求的核心逻辑，就全部定义在了当前类中，第二版本代码大家就全清楚了
 */
public abstract class NettyRemotingAbstract {


    private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_REMOTING_NAME);


    //以下两个信号量的具体使用方法和设计原理，请大家到本类的invoke0()方法中查看
    //单向请求信号量
    protected final Semaphore semaphoreOneway;

    //异步请求信号量
    protected final Semaphore semaphoreAsync;

    //专门处理NettyEvent事件的执行器
    protected final NettyEventExecutor nettyEventExecutor = new NettyEventExecutor();

    //存放RPC回调对象的集合
    protected List<RPCHook> rpcHooks = new ArrayList<>();

    //服务器是否正在关闭的标志
    protected AtomicBoolean isShuttingDown = new AtomicBoolean(false);

    //这个就是响应表，每发送一个请求就要创建一个ResponseFuture对象，用来接收与这个请求对应的响应
    //接收到响应并处理响应的时候，就会用到这个响应表，响应表的key就是请求和响应共享的唯一Id
    protected final ConcurrentMap<Integer /* opaque */, ResponseFuture> responseTable = new ConcurrentHashMap<>(256);

    //请求码和封装了与请求对应的请求处理器、执行器的Pair对象的映射表
    protected final HashMap<Integer/* request code */, Pair<NettyRequestProcessor, ExecutorService>> processorTable = new HashMap<>(64);

    //默认请求处理器和执行器的映射表
    protected Pair<NettyRequestProcessor, ExecutorService> defaultRequestProcessorPair;


    //初始化Netty的日志组件
    static {
        NettyLogger.initNettyLogger();
    }


    //构造方法
    public NettyRemotingAbstract(final int permitsOneway, final int permitsAsync) {
        this.semaphoreOneway = new Semaphore(permitsOneway, true);
        this.semaphoreAsync = new Semaphore(permitsAsync, true);
    }


    public abstract ChannelEventListener getChannelEventListener();


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/25
     * @方法描述：把NettyEvent事件对象添加到nettyEventExecutor执行器内部任务队列中的方法
     */
    public void putNettyEvent(final NettyEvent event) {
        this.nettyEventExecutor.putNettyEvent(event);
    }



    //注册RPC回调对象的方法
    public void registerRPCHook(RPCHook rpcHook) {
        if (rpcHook != null && !rpcHooks.contains(rpcHook)) {
            rpcHooks.add(rpcHook);
        }
    }


    //清空RPC回调对象方法
    public void clearRPCHook() {
        rpcHooks.clear();
    }


    //得到RPC回调对象的方法
    public List<RPCHook> getRPCHook() {
        return rpcHooks;
    }


    //得到用来执行回调方法的执行器
    public abstract ExecutorService getCallbackExecutor();

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/27
     * @方法描述：Namesrv处理接收到的消息的方法
     */
    public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) {
        //对消息进行判空处理
        if (msg != null) {
            //得到接收到的消息的类型
            switch (msg.getType()) {
                //判断是否为请求消息
                case REQUEST_COMMAND:
                    //处理请求
                    processRequestCommand(ctx, msg);
                    break;
                //判断是否为响应消息
                case RESPONSE_COMMAND:
                    //处理响应
                    processResponseCommand(ctx, msg);
                    break;
                default:
                    break;
            }
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/27
     * @方法描述:在RPC操作之前执行回调方法
     */
    protected void doBeforeRpcHooks(String addr, RemotingCommand request) {
        if (rpcHooks.size() > 0) {
            for (RPCHook rpcHook : rpcHooks) {
                rpcHook.doBeforeRequest(addr, request);
            }
        }
    }

   /**
    * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
    * @author：陈清风扬，个人微信号：chenqingfengyangjj。
    * @date:2024/12/27
    * @方法描述：执行RPC操作之后的回调方法
    */
    public void doAfterRpcHooks(String addr, RemotingCommand request, RemotingCommand response) {
        if (rpcHooks.size() > 0) {
            for (RPCHook rpcHook : rpcHooks) {
                rpcHook.doAfterResponse(addr, request, response);
            }
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/27
     * @方法描述：回复响应给客户端的方法
     */
    public static void writeResponse(Channel channel, RemotingCommand request, @Nullable RemotingCommand response) {
        writeResponse(channel, request, response, null);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/27
     * @方法描述：把响应回复给客户端的方法
     */
    public static void writeResponse(Channel channel, RemotingCommand request, @Nullable RemotingCommand response, Consumer<Future<?>> callback) {
        //对响应进行判空
        if (response == null) {
            return;
        }
        //设置与请求对应的唯一Id
        response.setOpaque(request.getOpaque());
        //设置消息类型为响应
        response.markResponseType();
        try {
            //把响应写到channel中
            channel.writeAndFlush(response).addListener((ChannelFutureListener) future -> {
                //响应发送后执行回调方法，判断响应是否发送成功，然后执行不同的操作
                if (future.isSuccess()) {
                    log.debug("Response[request code: {}, response code: {}, opaque: {}] is written to channel{}",
                            request.getCode(), response.getCode(), response.getOpaque(), channel);
                } else {
                    log.error("Failed to write response[request code: {}, response code: {}, opaque: {}] to channel{}",
                            request.getCode(), response.getCode(), response.getOpaque(), channel, future.cause());
                }
                if (callback != null) {
                    callback.accept(future);
                }
            });
        } catch (Throwable e) {
            log.error("process request over, but response failed", e);
            log.error(request.toString());
            log.error(response.toString());
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/27
     * @方法描述：处理请求的方法
     */
    public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
        //根据请求码从处理器表中得到封装了处理该请求处理器的Pair对象
        final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());
        //判断是否从处理器表中得到了对应的Pair对象
        final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessorPair : matched;
        //得到请求的唯一Id
        final int opaque = cmd.getOpaque();
        if (pair == null) {
            //如果没有找到处理该请求的处理器，那就直接回复给客户端一个不支持该请求的响应
            String error = " request type " + cmd.getCode() + " not supported";
            //创建响应对象，回复不支持该请求的响应码
            final RemotingCommand response =
                    RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
            response.setOpaque(opaque);
            //用channel回复响应
            writeResponse(ctx.channel(), cmd, response);
            log.error(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) + error);
            return;
        }
        //执行到这里意味着找到了处理该请求的处理器，但这个时候并不用直接处理请求，因为Namesrv要接收的请求太多了，可能有很多客户端都向Namesrv发送了同一类型的请求
        //直接就让该请求的处理器处理的话，显然有点忙不过来，所以这个时候就可以把处理该请求的操作封装成一个任务，然后把任务提交到请求处理器处理请求的执行器任务队列中
        //等待执行器依次处理即可，这里就是把处理请求的操作封装成了一个任务
        Runnable run = buildProcessRequestHandler(ctx, cmd, pair, opaque);
        //判断当前服务器是否正在关闭
        if (isShuttingDown.get()) {
            //如果服务器正在关闭，并且MQ的版本大于5.1.4
            if (cmd.getVersion() > MQVersion.Version.V5_1_4.ordinal()) {
                //那就回复客户端服务器正在关闭的响应
                final RemotingCommand response = RemotingCommand.createResponseCommand(ResponseCode.GO_AWAY,
                        "please go away");
                //设置响应Id
                response.setOpaque(opaque);
                //回复客户端响应
                writeResponse(ctx.channel(), cmd, response);
                return;
            }
        }
        //执行到这里还要再判断一下要处理该请求的请求处理器是否拒绝处理请求，在Namesrv模块使用的请求处理器都不会拒绝处理请求
        //rejectRequest()方法默认返回false
        if (pair.getObject1().rejectRequest()) {
            //如果处理器拒绝处理请求，那就返回客户端一个服务器系统繁忙的响应码
            final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                    "[REJECTREQUEST]system busy, start flow control for a while");
            //设置响应Id
            response.setOpaque(opaque);
            //回复客户端请求
            writeResponse(ctx.channel(), cmd, response);
            return;
        }
        try {
            //程序执行到这里意味着服务器状态正常，请求处理器也可以处理请求，这个时候就可以进一步封装异步任务
            //刚才我们已经把处理请求的操作封装成run这个任务了，现在要进一步对这个任务包装一下，把这个任务封装到RequestTask任务中
            //之所以这么做是因为RequestTask任务可以计算请求被处理的慢速时间
            final RequestTask requestTask = new RequestTask(run, ctx.channel(), cmd);
            //pair中存储了请求处理器和该请求处理器要处理请求的执行器，这里就是把任务对象提交给执行器的任务队列了
            pair.getObject2().submit(requestTask);
        } catch (RejectedExecutionException e) {
            //如果触发线程池的拒绝策略，说明线程池任务队列满了
            if ((System.currentTimeMillis() % 10000) == 0) {
                //记录一下告警日志
                log.warn(RemotingHelper.parseChannelRemoteAddr(ctx.channel())
                        + ", too many requests and system thread pool busy, RejectedExecutionException "
                        + pair.getObject2().toString()
                        + " request code: " + cmd.getCode());
            }
            //回复客户端系统繁忙的响应
            final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                    "[OVERLOAD]system busy, start flow control for a while");
            //设置响应Id
            response.setOpaque(opaque);
            //回复客户端响应
            writeResponse(ctx.channel(), cmd, response);
        } catch (Throwable e) {
            //忽略异常，这里在源码中并不是真的忽略异常
            //而是记录相关信息到性能检测组件中，但在我们的代码中还没有引入性能检测组件，所以我就直接把这里的代码注释了
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：把处理请求的操作封装成了任务的方法
     */
    private Runnable buildProcessRequestHandler(ChannelHandlerContext ctx, RemotingCommand cmd,
                                                Pair<NettyRequestProcessor, ExecutorService> pair, int opaque) {
        //直接返回一个Runnable对象
        return () -> {
            Exception exception = null;
            RemotingCommand response;
            try {
                //得到客户端的IP地址
                String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                try {
                    //在这里执行了RPC操作之前的回调方法
                    doBeforeRpcHooks(remoteAddr, cmd);
                } catch (AbortProcessException e) {
                    throw e;
                } catch (Exception e) {
                    exception = e;
                }
                if (exception == null) {
                    //判断程序执行过程中是否出现了异常，如果没有异常，则在这里获得请求处理器
                    //然后使用请求处理器处理接收到的请求
                    response = pair.getObject1().processRequest(ctx, cmd);
                } else {
                    //如果出现异常则回复客户端一个系统错误的响应码
                    response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_ERROR, null);
                }
                try {
                    //请求处理完毕之后，要回复给客户端响应了，在这里执行以下回复响应之前的RPC回调方法
                    doAfterRpcHooks(remoteAddr, cmd, response);
                } catch (AbortProcessException e) {
                    throw e;
                } catch (Exception e) {
                    exception = e;
                }
                if (exception != null) {
                    throw exception;
                }
                //在这里把响应回复给客户端
                writeResponse(ctx.channel(), cmd, response);
            } catch (AbortProcessException e) {
                response = RemotingCommand.createResponseCommand(e.getResponseCode(), e.getErrorMessage());
                response.setOpaque(opaque);
                writeResponse(ctx.channel(), cmd, response);
            } catch (Throwable e) {
                log.error("process request exception", e);
                log.error(cmd.toString());
                if (!cmd.isOnewayRPC()) {
                    response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_ERROR,
                            UtilAll.exceptionSimpleDesc(e));
                    response.setOpaque(opaque);
                    writeResponse(ctx.channel(), cmd, response);
                }
            }
        };
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：处理接收到的响应的方法
     */
    public void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd) {
        //先获得响应的唯一Id
        final int opaque = cmd.getOpaque();
        //从响应表中得到对应的ResponseFuture对象
        final ResponseFuture responseFuture = responseTable.get(opaque);
        if (responseFuture != null) {
            //如果ResponseFuture对象不为null，就把接收到的响应设置到ResponseFuture对象中
            //方便ResponseFuture对象内部的回调方法使用
            responseFuture.setResponseCommand(cmd);
            //到这时候响应也接收到了，也就可以把ResponseFuture对象从响应表中移除了，避免对与ResponseFuture对象对应的请求产生请求超时的误判
            responseTable.remove(opaque);
            //判断ResponseFuture对象中是否设置了回调方法
            if (responseFuture.getInvokeCallback() != null) {
                //如果设置了，这个时候就可以执行回调方法了
                executeInvokeCallback(responseFuture);
            } else {
                //如果没有设置回调方法，这里就把响应对象设置到ResponseFuture对象中即可
                //这里大家可能会疑惑为什么上面已经调用了setResponseCommand()方法把响应对象设置到ResponseFuture中
                //那这里为什么还要再设置一次呢？这里要结合ResponseFuture类的waitResponse()方法来理解
                //当ResponseFuture类的waitResponse()方法被调用时，会阻塞等待响应的线程，一旦调用了ResponseFuture类的putResponse()方法
                //线程就结束阻塞，开始继续执行程序，但是ResponseFuture类的setResponseCommand()方法并没有结束线程阻塞的功能
                //这就是ResponseFuture类的setResponseCommand()方法和putResponse()方法的区别，但是在源码中并没有真的用到
                //ResponseFuture类的putResponse()方法阻塞线程同步等待响应，所以大家对这个地方的内容简单了解一下就行
                responseFuture.putResponse(cmd);
                //释放ResponseFuture对象内部的信号量资源
                //表示一次请求响应全流程结束了
                responseFuture.release();
            }
        } else {
            //没有从响应表中找到对应的ResponseFuture对象，就记录告警日志，这时候很可能就是响应到达的太晚了，请求已经过期被丢弃了
            log.warn("receive response, cmd={}, but not matched any request, address={}", cmd, RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：在接收到响应之后，执行ResponseFuture对象中回调方法的方法
     */
    private void executeInvokeCallback(final ResponseFuture responseFuture) {
        //定义一个是否在当前线程执行该方法的标志，注意，这里的操作都是执行processMessageReceived()方法的线程执行的
        //而processMessageReceived()方法会在NettyServerHandler处理器的channelRead0()方法中被调用
        //而NettyServerHandler处理器的channelRead0()方法会被defaultEventExecutorGroup执行器的线程执行，所以这里大家要理清楚
        //执行当前方法的线程是NettyRemotingServer类的defaultEventExecutorGroup执行器
        boolean runInThisThread = false;
        //得到执行回调方法的执行器
        ExecutorService executor = this.getCallbackExecutor();
        //判断执行器是否不为null，以及是否还没有停止工作
        if (executor != null && !executor.isShutdown()) {
            try {
                //如果执行器状态正常，那就把执行回调方法的操作交给回调执行器执行，毕竟
                //NettyRemotingServer类的defaultEventExecutorGroup执行器还要赶快处理下一批数据，所以回调方法就交给其他的执行器异步执行
                executor.submit(() -> {
                    try {
                        //在这里执行回调方法
                        responseFuture.executeInvokeCallback();
                    } catch (Throwable e) {
                        //如果回调方法执行的过程中发生异常了，旧记录告警信息
                        log.warn("execute callback in executor exception, and callback throw", e);
                    } finally {
                        //回调方法执行完毕了，最后要把responseFuture中的信号量释放了，表示一次请求响应全流程结束了
                        responseFuture.release();
                    }
                });
            } catch (Exception e) {
                //如果在提交任务给回调执行器的时候出现异常了，提交任务失败，那就更新runInThisThread标志
                //表示回调方法在当前线程中执行，也就是NettyRemotingServer类的defaultEventExecutorGroup执行器执行该回调方法
                runInThisThread = true;
                log.warn("execute callback in executor exception, maybe executor busy", e);
            }
        } else {
            //如果回调执行器已经完毕了，那也更新runInThisThread标志
            runInThisThread = true;
        }
        //判断回调方法是否要在当前线程中执行
        if (runInThisThread) {
            try {
                //如果是的话就直接让当前线程执行回调方法
                responseFuture.executeInvokeCallback();
            } catch (Throwable e) {
                log.warn("executeInvokeCallback Exception", e);
            } finally {
                //把responseFuture中的信号量释放了，表示一次请求响应全流程结束了
                responseFuture.release();
            }
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：扫描响应表的方法，从响应表中清除请求超时的ResponseFuture对象
     */
    public void scanResponseTable() {
        //定义一个集合，这个集合用来存放需要清除的ResponseFuture对象
        final List<ResponseFuture> rfList = new LinkedList<>();
        //开始遍历响应表中的ResponseFuture对象
        Iterator<Map.Entry<Integer, ResponseFuture>> it = this.responseTable.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<Integer, ResponseFuture> next = it.next();
            //得到当前正在遍历的ResponseFuture对象
            ResponseFuture rep = next.getValue();
            //当前时间 > 请求开始发送的时间 + 请求最大超时时间 + 1000毫秒就意味着请求超时了
            //判断与该ResponseFuture对象对应的请求是否超时了，如果该请求及时接收到响应，那么与其对应的ResponseFuture对象就会从响应表中移除了
            //如果一直没有接收到响应，那在扫描响应表的时候，就会判断这个请求是否超时了，如果超时了就从响应表中清除该ResponseFuture对象
            if ((rep.getBeginTimestamp() + rep.getTimeoutMillis() + 1000) <= System.currentTimeMillis()) {
                //释放ResponseFuture对象中的信号量，该信号量一被释放，就意味着可以发送新的请求数量增加了
                rep.release();
                //从响应表中移除当前遍历的ResponseFuture对象
                it.remove();
                //把该对象添加到集合中
                rfList.add(rep);
                log.warn("remove timeout request, " + rep);
            }
        }
        for (ResponseFuture rf : rfList) {
            try {
                //执行要被释放的ResponseFuture对象中的回调方法
                executeInvokeCallback(rf);
            } catch (Throwable e) {
                log.warn("scanResponseTable, operationComplete Exception", e);
            }
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：服务器主动对外发送请求的方法，这个是同步发送请求的方法
     */
    public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
                                          final long timeoutMillis)
            throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
        try {
            //发送请求，并且指定了等待响应的超时时间，invokeImpl()方法会返回一个CompletableFuture对象，在指定的时间之后
            //CompletableFuture对象会通过thenApply方法得到响应结果，然后把响应结果返回给外层方法
            //在看完上面的代码之后，大家应该都清楚了响应对象RemotingCommand是怎么被设置到ResponseFuture对象中的
            return invokeImpl(channel, request, timeoutMillis).thenApply(ResponseFuture::getResponseCommand)
                    .get(timeoutMillis, TimeUnit.MILLISECONDS);
        } catch (ExecutionException e) {
            throw new RemotingSendRequestException(channel.remoteAddress().toString(), e.getCause());
        } catch (TimeoutException e) {
            throw new RemotingTimeoutException(channel.remoteAddress().toString(), timeoutMillis, e.getCause());
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：服务器主动对外发送请求的方法
     */
    public CompletableFuture<ResponseFuture> invokeImpl(final Channel channel, final RemotingCommand request,
                                                        final long timeoutMillis) {
        //得到客户端的IP地址
        String channelRemoteAddr = RemotingHelper.parseChannelRemoteAddr(channel);
        //执行RPC操作之前的回调方法
        doBeforeRpcHooks(channelRemoteAddr, request);
        //在这里把请求发送给客户端，invoke0()方法会返回一个CompletableFuture对象，并且当CompletableFuture对象Complete之后
        //就会执行RPC操作之后的回调方法
        return invoke0(channel, request, timeoutMillis).whenComplete((v, t) -> {
            if (t == null) {
                //再接收到响应后，执行RPC操作之后的回调方法
                doAfterRpcHooks(channelRemoteAddr, request, v.getResponseCommand());
            }
        });
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：发送请求的方法
     */
    protected CompletableFuture<ResponseFuture> invoke0(final Channel channel, final RemotingCommand request,
                                                        final long timeoutMillis) {
        //创建一个CompletableFuture对象，这个CompletableFuture对象用来封装一个ResponseFuture对象
        CompletableFuture<ResponseFuture> future = new CompletableFuture<>();
        //得到请求发送操作开始执行的时间
        long beginStartTime = System.currentTimeMillis();
        //得到请求的唯一标识
        final int opaque = request.getOpaque();
        //定义一个是否获得了信号量的标志
        boolean acquired;
        try {
            //尝试在指定时间获取异步发送请求的信号量指标，这里之所以设置这个信号量，是希望可以把服务器内发送请求的数量控制在一定范围之内
            //防止请求发送过多导致服务器内部资源被过度挤占，如果没有获取到信号量，就停止发送请求
            acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        } catch (Throwable t) {
            //没有获取到信号量则直接返回CompletableFuture，并且设置异常到CompletableFuture中
            future.completeExceptionally(t);
            return future;
        }
        if (acquired) {
            //执行到这里就意味着成功获得到信号量了，也就意味着可以成功发送本次请求
            //那就把这个信号量封装到SemaphoreReleaseOnlyOnce对象中，这么做的用意也很简单，就是防止同意信号量被多次释放
            //这次申请到的信号量只被释放一次
            final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreAsync);
            //得到从操作开始到当前的消耗时间，获取信号量也许会消耗一部分时间
            long costTime = System.currentTimeMillis() - beginStartTime;
            //得到的这个操作消耗时间和请求超时时间对比一下，如果请求超时时间小于操作消耗时间，说明执行发送请求的操作早就超过请求超时时间了
            //这时候也就没必要继续发送请求了
            if (timeoutMillis < costTime) {
                //释放信号量
                once.release();
                //设置异常并返回CompletableFuture对象
                future.completeExceptionally(new RemotingTimeoutException("invokeAsyncImpl call timeout"));
                return future;
            }
            //执行到这里意味着信号量也获取到了，请求也没有超时，接下来就可以执行发送请求的操作了
            //这里创建的这个原子引用对象是用来持有ResponseFuture对象的
            AtomicReference<ResponseFuture> responseFutureReference = new AtomicReference<>();
            //创建一个ResponseFuture对象，并且设置了请求超时时间，以及响应回调对象，信号量对象
            final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, request, timeoutMillis - costTime,
                    new InvokeCallback() {

                        //该方法会在请求——响应全流程结束之后被回调
                        @Override
                        public void operationComplete(ResponseFuture responseFuture) {
                            //该方法逻辑可以自由实现
                        }

                        //该方法会在该请求成功接收到响应后被回调
                        @Override
                        public void operationSucceed(RemotingCommand response) {
                            //这个future是最开始创建的CompletableFuture对象
                            //responseFutureReference是刚才创建的持有了ResponseFuture对象的原子引用对象，而ResponseFuture对象在请求发送之前就已经设置到responseFutureReference源自引用对象中了
                            //这里的意思就是在成功接收到响应之后，响应已经设置到ResponseFuture对象中了，那我们就可以从responseFutureReference原子引用对象中把封装了响应的ResponseFuture对象
                            //设置到CompletableFuture对象中，这样一来，CompletableFuture的回调方法就可以直接得到响应对象了，然后执行CompletableFuture的回调方法即可
                            //这个逻辑有一点绕，大家要仔细梳理一下
                            future.complete(responseFutureReference.get());
                        }

                        //该方法会在获取响应失败时被回调
                        @Override
                        public void operationFail(Throwable throwable) {
                            //设置异常到CompletableFuture对象中
                            future.completeExceptionally(throwable);
                        }//注意，这里把信号量对象设置到ResponseFuture对象中了
                    }, once);
            //在发送请求之前把把ResponseFuture对象交给responseFutureReference原子引用对象了
            responseFutureReference.set(responseFuture);
            //把ResponseFuture存储到响应表中
            this.responseTable.put(opaque, responseFuture);
            try {
                //在这里把请求发送出去了
                channel.writeAndFlush(request).addListener((ChannelFutureListener) f -> {
                    //判断请求是否成功发送
                    if (f.isSuccess()) {
                        //发送成功则给ResponseFuture对象设置请求成功发送标志
                        responseFuture.setSendRequestOK(true);
                        return;
                    }
                    //请求发送失败则执行该方法中的操作，无法就是从响应表中移除对应的ResponseFuture对象，释放信号量资源等等
                    requestFail(opaque);
                    //记录请求发送失败的告警日志
                    log.warn("send a request command to channel <{}> failed.", RemotingHelper.parseChannelRemoteAddr(channel));
                });
                return future;
            } catch (Exception e) {
                //请求发送过程中出现异常则从响应表中移除对应的ResponseFuture响应对象
                responseTable.remove(opaque);
                //释放信号量资源
                responseFuture.release();
                log.warn("send a request command to channel <" + RemotingHelper.parseChannelRemoteAddr(channel) + "> Exception", e);
                future.completeExceptionally(new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e));
                return future;
            }
        } else {
            //执行到这里意味着没有成功获得到信号量，判断是否设置了请求超时时间
            if (timeoutMillis <= 0) {
                //如果没有设置就返回一个请求操作执行太快的异常信息
                future.completeExceptionally(new RemotingTooMuchRequestException("invokeAsyncImpl invoke too fast"));
            } else {
                //如果设置了请求超时时间，就返回获得信号量超时异常信息
                String info = String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
                                timeoutMillis,
                                this.semaphoreAsync.getQueueLength(),
                                this.semaphoreAsync.availablePermits());
                log.warn(info);
                future.completeExceptionally(new RemotingTimeoutException(info));
            }
            return future;
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：异步发送请求的方法，该方法逻辑和之前的方法类似，我就不添加详细注释了
     */
    public void invokeAsyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis,
                                final InvokeCallback invokeCallback) {
        invokeImpl(channel, request, timeoutMillis)
                .whenComplete((v, t) -> {
                    if (t == null) {
                        invokeCallback.operationComplete(v);
                    } else {
                        ResponseFuture responseFuture = new ResponseFuture(channel, request.getOpaque(), request, timeoutMillis, null, null);
                        responseFuture.setCause(t);
                        invokeCallback.operationComplete(responseFuture);
                    }
                })
                .thenAccept(responseFuture -> invokeCallback.operationSucceed(responseFuture.getResponseCommand()))
                .exceptionally(t -> {
                    invokeCallback.operationFail(t);
                    return null;
                });
    }


   /**
    * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
    * @author：陈清风扬，个人微信号：chenqingfengyangjj。
    * @date:2024/12/28
    * @方法描述：请求发送失败要执行的方法，就是把请求的响应对象从响应表中移除，然后释放信号量
    */
    private void requestFail(final int opaque) {
        //把对应的ResponseFuture对象从响应表中移除
        ResponseFuture responseFuture = responseTable.remove(opaque);
        if (responseFuture != null) {
            //设置请求发送失败标志
            responseFuture.setSendRequestOK(false);
            //设置接收到的响应为null
            responseFuture.putResponse(null);
            try {
                //执行回调方法
                executeInvokeCallback(responseFuture);
            } catch (Throwable e) {
                log.warn("execute callback in requestFail, and callback throw", e);
            } finally {
                //释放信号量资源
                responseFuture.release();
            }
        }
    }


    protected void failFast(final Channel channel) {
        for (Map.Entry<Integer, ResponseFuture> entry : responseTable.entrySet()) {
            if (entry.getValue().getChannel() == channel) {
                Integer opaque = entry.getKey();
                if (opaque != null) {
                    requestFail(opaque);
                }
            }
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：发送单项请求的方法，不必关心响应结果
     */
    public void invokeOnewayImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis)
            throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
        //设置请求类型为单向请求，不需要关心响应结果
        request.markOnewayRPC();
        //尝试获取信号量
        boolean acquired = this.semaphoreOneway.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        if (acquired) {
            final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreOneway);
            try {
                //成功获得信号量后发送请求
                channel.writeAndFlush(request).addListener((ChannelFutureListener) f -> {
                    once.release();
                    if (!f.isSuccess()) {
                        log.warn("send a request command to channel <" + channel.remoteAddress() + "> failed.");
                    }
                });
            } catch (Exception e) {
                once.release();
                log.warn("write send a request command to channel <" + channel.remoteAddress() + "> failed.");
                throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
            }
        } else {
            //以下是信号量获取失败的操作
            if (timeoutMillis <= 0) {
                throw new RemotingTooMuchRequestException("invokeOnewayImpl invoke too fast");
            } else {
                String info = String.format("invokeOnewayImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreOnewayValue: %d", timeoutMillis, this.semaphoreOneway.getQueueLength(), this.semaphoreOneway.availablePermits()
                );
                log.warn(info);
                throw new RemotingTimeoutException(info);
            }
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/25
     * @方法描述：该类的对象就是一个单线程执行器，这个执行器就是专门用来处理NettyEvent事件对象的
     */
    class NettyEventExecutor extends ServiceThread {

        //存储NettyEvent事件对象的任务队列
        private final LinkedBlockingQueue<NettyEvent> eventQueue = new LinkedBlockingQueue<>();


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @date:2024/12/25
         * @方法描述：把NettyEvent事件对象存放到任务队列的方法，该方法会在NettyConnectManageHandler类中被调用
         */
        public void putNettyEvent(final NettyEvent event) {
            int currentSize = this.eventQueue.size();
            int maxSize = 10000;
            //当前队列还有容量
            if (currentSize <= maxSize) {
                //直接把NettyEvent事件对象添加到任务队列中
                this.eventQueue.add(event);
            } else {
                //超过最大容量了则丢掉事件对象，记录告警信息
                log.warn("event queue size [{}] over the limit [{}], so drop this event {}", currentSize, maxSize, event.toString());
            }
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @date:2024/12/25
         * @方法描述：单线程执行器要执行的Runnable任务
         */
        @Override
        public void run() {
            log.info(this.getServiceName() + " service started");

            //得到客户端channel监听器，这里得到的就是NamesrvController中的brokerHousekeepingService成员变量
            final ChannelEventListener listener = NettyRemotingAbstract.this.getChannelEventListener();

            //在循环中不断处理任务队列中的事件
            while (!this.isStopped()) {
                try {
                    //有限时地从队列中获取事件
                    NettyEvent event = this.eventQueue.poll(3000, TimeUnit.MILLISECONDS);
                    //如果得到了NettyEvent事件对象，那就根据事件对象的类型，调用监听器对象的方法
                    if (event != null && listener != null) {
                        switch (event.getType()) {
                            //如果是空闲事件，那就调用监听器的onChannelIdle方法，以下逻辑都是如此，我就不再重复添加注释了
                            case IDLE:
                                listener.onChannelIdle(event.getRemoteAddr(), event.getChannel());
                                break;
                            case CLOSE:
                                listener.onChannelClose(event.getRemoteAddr(), event.getChannel());
                                break;
                            case CONNECT:
                                listener.onChannelConnect(event.getRemoteAddr(), event.getChannel());
                                break;
                            case EXCEPTION:
                                listener.onChannelException(event.getRemoteAddr(), event.getChannel());
                                break;
                            case ACTIVE:
                                listener.onChannelActive(event.getRemoteAddr(), event.getChannel());
                                break;
                            default:
                                break;
                        }
                    }
                } catch (Exception e) {
                    log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }
            log.info(this.getServiceName() + " service end");
        }



        @Override
        public String getServiceName() {
            return NettyEventExecutor.class.getSimpleName();
        }
    }
}
